package grabbitmq

import (
	"context"
	"errors"
	"fmt"
	"go.uber.org/zap"
	"sync"
	"time"
	"wzgames/pkg/logger"

	amqp "github.com/rabbitmq/amqp091-go"
)

type Producer struct {
	name string

	mq *MQ

	chPool *sync.Pool

	publishMutex sync.RWMutex

	mutex sync.RWMutex

	ch *amqp.Channel

	exchangeBinds []*ExchangeBinds

	enableConfirm bool
	// listen publish confirm
	confirmC chan amqp.Confirmation
	// confirm result
	confirm *confirmHelper

	// listen chanel close
	closeC chan *amqp.Error
	stopC  chan struct{}

	state uint8
}

func newProducer(name string, mq *MQ) *Producer {
	return &Producer{
		name: name,
		mq:   mq,
		chPool: &sync.Pool{
			New: func() interface{} { return make(chan bool, 1) },
		},
		state: StateClosed,
	}
}

func (p *Producer) Name() string {
	return p.name
}

func (p *Producer) Confirm(enable bool) *Producer {
	p.mutex.Lock()
	p.enableConfirm = enable
	p.mutex.Unlock()
	return p
}

func (p *Producer) SetExchangeBinds(eb []*ExchangeBinds) *Producer {
	p.mutex.Lock()
	if p.state != StateOpened {
		p.exchangeBinds = eb
	}
	p.mutex.Unlock()
	return p
}

func (p *Producer) IsOpen() bool {
	p.mutex.Lock()
	defer p.mutex.Unlock()
	return p.state == StateOpened
}

func (p *Producer) Open() error {
	p.mutex.Lock()
	defer p.mutex.Unlock()

	if p.mq == nil {
		return errors.New("MQ: Bad producer")
	}
	if len(p.exchangeBinds) <= 0 {
		return errors.New("MQ: No exchangeBinds found. You should SetExchangeBinds before open.")
	}
	if p.state == StateOpened {
		return errors.New("MQ: Producer had been opened")
	}

	ch, err := p.mq.channel()
	if err != nil {
		return fmt.Errorf("MQ: Create channel failed, %v", err)
	}
	if err = applyExchangeBinds(ch, p.exchangeBinds); err != nil {
		ch.Close()
		return err
	}

	p.ch = ch
	p.state = StateOpened

	// Confirm
	if p.enableConfirm {
		p.confirmC = make(chan amqp.Confirmation, 1)
		p.ch.Confirm(false)
		p.ch.NotifyPublish(p.confirmC)
		if p.confirm == nil {
			p.confirm = newConfirmHelper()
		} else {
			p.confirm.Reset()
		}

		go p.listenConfirm()
	}

	// Keepalive
	if true {
		p.stopC = make(chan struct{})
		p.closeC = make(chan *amqp.Error, 1)
		p.ch.NotifyClose(p.closeC)

		go p.keepalive()
	}

	return nil
}

func (p *Producer) Close() {
	p.mutex.Lock()
	defer p.mutex.Unlock()

	select {
	case <-p.stopC:
		// had been closed
	default:
		close(p.stopC)
	}
}

func (p *Producer) PublishT(exchange, routeKey string, data []byte) error {
	return p.publish(exchange, routeKey, Transient, data)
}

func (p *Producer) PublishP(exchange, routeKey string, data []byte) error {
	return p.publish(exchange, routeKey, Persistent, data)
}

func (p *Producer) publish(exchange, routeKey string, delMode uint8, data []byte) error {
	if data == nil {
		return errors.New("MQ: Nil publish msg")
	}

	if st := p.State(); st != StateOpened {
		return fmt.Errorf("MQ: Producer unopened, now state is %d", p.state)
	}

	pub := amqp.Publishing{
		//ContentType:     msg.ContentType,
		//ContentEncoding: msg.ContentEncoding,
		DeliveryMode: delMode,
		Body:         data,
	}

	if p.enableConfirm == false {
		return p.ch.PublishWithContext(context.Background(), exchange, routeKey, false, false, pub)
	}

	// confirm mode
	p.publishMutex.Lock()
	if err := p.ch.PublishWithContext(context.Background(), exchange, routeKey, false, false, pub); err != nil {
		p.publishMutex.Unlock()
		return fmt.Errorf("MQ: Producer publish failed, %v", err)
	}
	ch := p.chPool.Get().(chan bool)
	p.confirm.Listen(ch)
	p.publishMutex.Unlock()

	ack, ok := <-ch
	p.chPool.Put(ch)
	if !ack || !ok {
		return fmt.Errorf("MQ: Producer publish failed, confirm ack is false. ack:%t, ok:%t", ack, ok)
	}
	return nil
}

func (p *Producer) State() uint8 {
	p.mutex.RLock()
	defer p.mutex.RUnlock()
	return p.state
}

func (p *Producer) keepalive() {
	select {
	case <-p.stopC:
		// 正常关闭
		logger.Warn("[RabbitMQ] Producer shutdown normally.", zap.String("ProducerName", p.name))
		p.mutex.Lock()
		p.ch.Close()
		p.state = StateClosed
		p.mutex.Unlock()

	case err := <-p.closeC:
		if err == nil {
			logger.Error("[RabbitMQ] Producer channel was closed, but Error detail is nil", zap.String("ProducerName", p.name))
		} else {
			logger.Error("[RabbitMQ] Producer channel was closed", zap.String("ProducerName", p.name), zap.Int("ErrCode", err.Code), zap.Error(err))
		}

		// channel closed exception
		p.mutex.Lock()
		p.state = StateReopening
		p.mutex.Unlock()

		maxRetry := 99999999
		for i := 0; i < maxRetry; i++ {
			time.Sleep(time.Second)
			if p.mq.State() != StateOpened {
				//glog.Warn("MQ: Producer try to recover channel, but mq's state != StateOpened", log.String("ProducerName", p.name), log.Int("Times", i+1))
				continue
			}
			if e := p.Open(); e != nil {
				//glog.Warn("MQ: Producer recover channel failed", log.String("ProducerName", p.name), log.Int("Times", i+1), log.Err(e))
				continue
			}
			return
		}
		logger.Error("[RabbitMQ] Producer try to recover channel over maxRetry and exit", zap.String("ProducerName", p.name), zap.Int("Times", maxRetry))
	}
}

func (p *Producer) listenConfirm() {
	for c := range p.confirmC {
		go p.confirm.Callback(c.DeliveryTag, c.Ack)
	}
}

func applyExchangeBinds(ch *amqp.Channel, exchangeBinds []*ExchangeBinds) (err error) {
	if ch == nil {
		return errors.New("MQ: Nil producer channel")
	}
	if len(exchangeBinds) <= 0 {
		return errors.New("MQ: Empty exchangeBinds")
	}

	for _, eb := range exchangeBinds {
		if eb.Exch == nil {
			return errors.New("MQ: Nil exchange found.")
		}
		// declare exchange
		ex := eb.Exch
		if err = ch.ExchangeDeclare(ex.Name, ex.Kind, ex.Durable, ex.AutoDelete, ex.Internal, ex.NoWait, ex.Args); err != nil {
			return fmt.Errorf("MQ: Declare exchange(%s) failed, %v", ex.Name, err)
		}
		if len(eb.Bindings) <= 0 {
			//return fmt.Errorf("MQ: No bindings queue found for exchange(%s)", eb.Exch.Name)
			return nil
		}
		// declare and bind queues
		for _, b := range eb.Bindings {
			if b == nil {
				return fmt.Errorf("MQ: Nil binding found, exchange:%s", ex.Name)
			}
			if len(b.Queues) <= 0 {
				return fmt.Errorf("MQ: No queues found for exchange(%s)", ex.Name)
			}
			for _, q := range b.Queues {
				if q == nil {
					return fmt.Errorf("MQ: Nil queue found, exchange:%s", ex.Name)
				}
				if _, err = ch.QueueDeclare(q.Name, q.Durable, q.AutoDelete, q.Exclusive, q.NoWait, q.Args); err != nil {
					return fmt.Errorf("MQ: Declare queue(%s) failed, %v", q.Name, err)
				}
				if err = ch.QueueBind(q.Name, b.RouteKey, ex.Name, b.NoWait, b.Args); err != nil {
					return fmt.Errorf("MQ: Bind exchange(%s) <--> queue(%s) failed, %v", ex.Name, q.Name, err)
				}
			}
		}
	}
	return nil
}

type confirmHelper struct {
	mutex     sync.RWMutex
	listeners map[uint64]chan<- bool
	count     uint64
}

func newConfirmHelper() *confirmHelper {
	h := confirmHelper{}
	return h.Reset()
}

func (h *confirmHelper) Reset() *confirmHelper {
	h.mutex.Lock()
	defer h.mutex.Unlock()

	for _, ch := range h.listeners {
		close(ch)
	}

	// Reset
	h.count = uint64(0)
	h.listeners = make(map[uint64]chan<- bool)
	return h
}

func (h *confirmHelper) Listen(ch chan<- bool) {
	h.mutex.Lock()
	h.count++
	h.listeners[h.count] = ch
	h.mutex.Unlock()
}

func (h *confirmHelper) Callback(idx uint64, ack bool) {
	h.mutex.Lock()
	ch := h.listeners[idx]
	delete(h.listeners, idx)
	h.mutex.Unlock()
	ch <- ack
}
